from openmind_hub import create_repo,delete_repo,create_branch,delete_branch
import unittest
import time
from openmind_hub import RepositoryNotFoundError, GatedRepoError
from openmind_hub.plugins.openmind.utils._error import BadRequestError

token = '123456789()'
repo_id = "guoxiaozhen2/model_test2024"
license = "mit"
branch = "v3"

class TestOpenmindHub(unittest.TestCase):
    def test_01_create_modelbranch_success(self):
        try:
            create_repo(
                repo_id=repo_id,
                private=True,
                desc="描述内容",
                license=license,
                token=token,
                exist_ok=True,
                fullname=None
            )
            time.sleep(1)
            create_branch(
                repo_id=repo_id,
                branch=branch,
                revision="main",
                token=token,
                repo_type="model",
                exist_ok=False
            )
            time.sleep(5)
        except:
            raise Exception("创建仓库分支有报错")
        finally:
            delete_branch(
                repo_id=repo_id,
                branch=branch,
                token=token,
                repo_type="model",
            )
            delete_repo(
                repo_id=repo_id,
                token=token,
                repo_type="model",
                missing_ok=False
            )
    def test_02_create_modelbranch_duplicate_existok(self):
        try:
            create_repo(
                repo_id=repo_id,
                private=False,
                desc="描述内容",
                license=license,
                token=token,
                repo_type="model",
                exist_ok=True,
                fullname=None
            )
            time.sleep(1)
            create_branch(
                repo_id=repo_id,
                branch=branch,
                revision="main",
                token=token,
                repo_type="model",
                exist_ok=True
            )
            time.sleep(5)
            create_branch(
                repo_id=repo_id,
                branch=branch,
                revision="main",
                token=token,
                repo_type="model",
                exist_ok=True
            )
        except:
            raise Exception("exist_ok=True未生效")
        finally:
            delete_branch(
                repo_id=repo_id,
                branch=branch,
                token=token,
                repo_type="model",
            )
            delete_repo(
                repo_id=repo_id,
                token=token,
                repo_type="model",
                missing_ok=False
            )
    def test_03_create_modelbranch_duplicate_existnotok(self):
        try:
            create_repo(
                repo_id=repo_id,
                private=False,
                desc="描述内容",
                license=license,
                token=token,
                repo_type="model",
                exist_ok=True,
                fullname=None
            )
            time.sleep(1)
            create_branch(
                repo_id=repo_id,
                branch=branch,
                revision="main",
                token=token,
                repo_type="model",
                exist_ok=True
            )
            time.sleep(5)
            create_branch(
                repo_id=repo_id,
                branch=branch,
                revision="main",
                token=token,
                repo_type="model",
                exist_ok=False
            )
            raise Exception("exist_ok=False未生效")
        except BadRequestError:
            delete_branch(
                repo_id=repo_id,
                branch=branch,
                token=token,
                repo_type="model",
            )
            delete_repo(
                repo_id=repo_id,
                token=token,
                repo_type="model",
                missing_ok=False
            )
    def test_04_create_modelbranch_noexistrepoid(self):
        try:
            create_branch(
                repo_id=repo_id+"a",
                branch="test",
                revision="main",
                token=token,
                repo_type="model",
                exist_ok=False
            )
            raise Exception("不存在的仓库创建分支未报错")
        except RepositoryNotFoundError:
            pass
    def test_05_create_modelbranch_wrong_token(self):
        try:
            create_branch(
                repo_id=repo_id,
                branch="test",
                revision="main",
                token=token+"a",
                repo_type="model",
                exist_ok=False
            )
            raise Exception("token未校验")
        except GatedRepoError:
            pass
    def test_06_delete_modelbranch_success(self):
        try:
            create_repo(
                repo_id=repo_id,
                private=False,
                desc="描述内容",
                license=license,
                token=token,
                repo_type="model",
                exist_ok=True,
                fullname=None
            )
            time.sleep(1)
            create_branch(
                repo_id=repo_id,
                branch=branch,
                revision="main",
                token=token,
                repo_type="model",
                exist_ok=True
            )
            time.sleep(5)
            delete_branch(
                repo_id=repo_id,
                branch=branch,
                token=token,
                repo_type="model",
            )
        except:
            raise Exception("删除仓库分支有报错")
        finally:
            delete_repo(
                repo_id=repo_id,
                token=token,
                repo_type="model",
                missing_ok=False
            )
    def test_07_create_modelbranch_no_repo_type(self):
        try:
            create_repo(
                repo_id=repo_id,
                private=False,
                desc="描述内容",
                license=license,
                token=token,
                repo_type="model",
                exist_ok=True,
                fullname=None
            )
            time.sleep(1)
            create_branch(
                repo_id=repo_id,
                branch=branch,
                revision="main",
                token=token,
                exist_ok=False
            )
            time.sleep(3)
        except:
            raise Exception("仓库类型未添加默认值")
        else:
            delete_branch(
                repo_id=repo_id,
                branch=branch,
                token=token,
                repo_type="model",
            )
            delete_repo(
                repo_id=repo_id,
                token=token,
                repo_type="model",
                missing_ok=False
            )
    def test_08_delete_modelbranch_wrongtoken(self):
        try:
            create_repo(
                repo_id=repo_id,
                private=False,
                desc="描述内容",
                license=license,
                token=token,
                repo_type="model",
                exist_ok=True,
                fullname=None
            )
            time.sleep(1)
            create_branch(
                repo_id=repo_id,
                branch=branch,
                revision="main",
                token=token,
                exist_ok=False
            )
            time.sleep(5)
            delete_branch(
                repo_id=repo_id,
                branch=branch,
                token=token+"a",
                repo_type="model",
            )
            raise Exception("missing_ok=True未生效")
        except GatedRepoError:
            delete_branch(
                repo_id=repo_id,
                branch=branch,
                token=token,
                repo_type="model",
            )
            delete_repo(
                repo_id=repo_id,
                token=token,
                repo_type="model",
                missing_ok=False
            )
    def test_09_delete_modelbranch_noexistrepoid(self):
        try:
            delete_branch(
                repo_id=repo_id+"a",
                branch=branch,
                token=token,
                repo_type="model",
            )
            raise Exception("删除不存在的仓库分支未报错")
        except RepositoryNotFoundError:
            pass
    def test_10_create_spacebranch_success(self):
        space_sdk = "Gradio"
        space_hardware = "CPU basic 2 vCPU · 16GB · FREE"
        space_image = "openeuler-python3.9-mindspore2.3.1-openmind0.8.0"
        try:
            create_repo(
                repo_id=repo_id,
                private=True,
                desc="描述内容",
                license=license,
                token=token,
                repo_type="space",
                exist_ok=False,
                fullname=None,
                space_sdk=space_sdk,
                space_hardware=space_hardware,
                space_image=space_image,
            )
            time.sleep(3)
            create_branch(
                repo_id=repo_id,
                branch="test",
                revision="main",
                token=token,
                repo_type="space",
                exist_ok=True
            )
            time.sleep(5)
        except:
            raise Exception("创建仓库分支有报错")
        finally:
            delete_branch(
                repo_id=repo_id,
                branch="test",
                token=token,
                repo_type="space",
            )
            delete_repo(
                repo_id=repo_id,
                token=token,
                repo_type="space",
                missing_ok=True
            )
    def test_11_delete_spacebranch_success(self):
        space_sdk = "Gradio"
        space_hardware = "CPU basic 2 vCPU · 16GB · FREE"
        space_image = "openeuler-python3.9-mindspore2.3.1-openmind0.8.0"
        try:
            create_repo(
                repo_id=repo_id,
                private=True,
                desc="描述内容",
                license=license,
                token=token,
                repo_type="space",
                exist_ok=False,
                fullname=None,
                space_sdk=space_sdk,
                space_hardware=space_hardware,
                space_image=space_image,
            )
            time.sleep(3)
            create_branch(
                repo_id=repo_id,
                branch="test",
                revision="main",
                token=token,
                repo_type="space",
                exist_ok=True
            )
            time.sleep(5)
            delete_branch(
                repo_id=repo_id,
                branch="test",
                token=token,
                repo_type="space",
            )
        except:
            raise Exception("删除仓库分支有报错")
        finally:
            delete_repo(
                repo_id=repo_id,
                token=token,
                repo_type="space",
                missing_ok=True
            )
    def test_12_create_datasetbranch_success(self):
        try:
            create_repo(
                repo_id=repo_id,
                private=False,
                desc="描述内容",
                license=license,
                token=token,
                repo_type="dataset",
                exist_ok=True,
                fullname=None
            )
            time.sleep(3)
            create_branch(
                repo_id=repo_id,
                branch="test",
                revision="main",
                token=token,
                repo_type="dataset",
                exist_ok=True
            )
            time.sleep(3)
        except:
            raise Exception("创建仓库分支有报错")
        finally:
            delete_branch(
                repo_id=repo_id,
                branch="test",
                token=token,
                repo_type="dataset",
            )
            delete_repo(
                repo_id=repo_id,
                token=token,
                repo_type="dataset",
                missing_ok=True
            )
    def test_13_delete_datasetbranch_success(self):
        try:
            create_repo(
                repo_id=repo_id,
                private=False,
                desc="描述内容",
                license=license,
                token=token,
                repo_type="dataset",
                exist_ok=True,
                fullname=None
            )
            time.sleep(3)
            create_branch(
                repo_id=repo_id,
                branch="test",
                revision="main",
                token=token,
                repo_type="dataset",
                exist_ok=True
            )
            time.sleep(3)
            delete_branch(
                repo_id=repo_id,
                branch="test",
                token=token,
                repo_type="dataset",
            )
        except:
            raise Exception("删除仓库分支有报错")
        finally:
            delete_repo(
                repo_id=repo_id,
                token=token,
                repo_type="dataset",
                missing_ok=True
            )


if __name__ == '__main__':
    unittest.main()